package com.atguigu.flink.day03;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

/**
 * @author Felix
 * @date 2024/8/12
 * 该案例演示了自定义数据源Source
 */
public class Flink08_Source_Custom {
    public static void main(String[] args) throws Exception {
        //TODO 1.指定流处理环境
        Configuration conf = new Configuration();
        conf.set(RestOptions.PORT,8088);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        //StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        //TODO 2.自定义数据源
        DataStreamSource<String> ds = env.addSource(new SourceFunction<String>() {
            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                for (int i = 0; i < 100000000; i++) {
                    ctx.collect("数据" + i);
                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {
                System.out.println("~~~cancel被执行了~~~");
            }
        });
        //env.fromSource();
        //TODO 3.打印
        ds.print();
        //TODO 4.提交作业
        env.execute();
    }
}
